// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//
#pragma once

#include <pollux/core/plan_node.h>
#include <pollux/exec/operator.h>
#include <pollux/exec/operator_utils.h>
#include <pollux/expression/expr.h>

namespace kumo::pollux::exec {
    class FilterProject : public Operator {
    public:
        FilterProject(
            int32_t operatorId,
            DriverCtx *driverCtx,
            const std::shared_ptr<const core::FilterNode> &filter,
            const std::shared_ptr<const core::ProjectNode> &project);

        bool isFilter() const override {
            return true;
        }

        bool preservesOrder() const override {
            return true;
        }

        bool needsInput() const override {
            return !input_;
        }

        void addInput(RowVectorPtr input) override;

        RowVectorPtr getOutput() override;

        BlockingReason isBlocked(ContinueFuture * /* unused */) override {
            return BlockingReason::kNotBlocked;
        }

        bool isFinished() override;

        void close() override {
            Operator::close();
            if (exprs_ != nullptr) {
                exprs_->clear();
            } else {
                POLLUX_CHECK(!initialized_);
            }
        }

        /// Data for accelerator conversion.
        struct Export {
            const ExprSet *exprs;
            bool hasFilter;
            const std::vector<IdentityProjection> *resultProjections;
        };

        Export exprsAndProjection() const {
            return Export{exprs_.get(), hasFilter_, &resultProjections_};
        }

        void initialize() override;

    private:
        // Tests if 'numProcessedRows_' equals to the length of input_ and clears
        // outstanding references to input_ if done. Returns true if getOutput
        // should return nullptr.
        bool allInputProcessed();

        // Evaluate filter on all rows. Return number of rows that passed the filter.
        // Populate filterEvalCtx_.selectedBits and selectedIndices with the indices
        // of the passing rows if only some rows pass the filter. If all or no rows
        // passed the filter filterEvalCtx_.selectedBits and selectedIndices are not
        // updated.
        vector_size_t filter(EvalCtx &evalCtx, const SelectivityVector &allRows);

        // Evaluate projections on the specified rows and return the results.
        // pre-condition: !isIdentityProjection_
        std::vector<VectorPtr> project(
            const SelectivityVector &rows,
            EvalCtx &evalCtx);

        // If true exprs_[0] is a filter and the other expressions are projections
        const bool hasFilter_{false};

        // Cached filter and project node for lazy initialization. After
        // initialization, they will be reset, and initialized_ will be set to true.
        std::shared_ptr<const core::ProjectNode> project_;
        std::shared_ptr<const core::FilterNode> filter_;
        bool initialized_{false};

        std::unique_ptr<ExprSet> exprs_;
        int32_t numExprs_;

        FilterEvalCtx filterEvalCtx_;

        vector_size_t numProcessedInputRows_{0};

        // Indices for fields/input columns that are both an identity projection and
        // are referenced by either a filter or project expression. This is used to
        // identify fields that need to be preloaded before evaluating filters or
        // projections.
        // Consider projection with 2 expressions: f(c0) AND g(c1), c1
        // If c1 is a LazyVector and f(c0) AND g(c1) expression is evaluated first, it
        // will load c1 only for rows where f(c0) is true. However, c1 identity
        // projection needs all rows.
        std::vector<column_index_t> multiplyReferencedFieldIndices_;
    };
} // namespace kumo::pollux::exec
